package com.codejiwei.flink.table.redis

import com.codejiwei.flink.table.redis.source.RedisDynamicTableSource
import org.apache.flink.configuration.ConfigOption
import org.apache.flink.table.connector.sink.DynamicTableSink
import org.apache.flink.table.connector.source.DynamicTableSource
import org.apache.flink.table.factories.{DeserializationFormatFactory, DynamicTableFactory, DynamicTableSinkFactory, DynamicTableSourceFactory, FactoryUtil}
import com.codejiwei.flink.table.redis.option.RedisOptions._
import com.codejiwei.flink.table.redis.option.RtsqlRedisOptions.{BNS, FILTER, PASSWORD, UPDATE_MODE}
import org.apache.flink.api.common.serialization.DeserializationSchema
import org.apache.flink.table.connector.format.DecodingFormat
import org.apache.flink.table.data.RowData

import java.util

/**
 * Author: codejiwei
 * Date: 2022/7/7
 * Desc:  
 * */
class RedisDynamicTableFactory extends DynamicTableSourceFactory with DynamicTableSinkFactory{
  //匹配connector = "redis"
  override def factoryIdentifier(): String = "redis"

  override def requiredOptions(): util.Set[ConfigOption[_]] = {
    val options = new util.HashSet[ConfigOption[_]]()
    options.add(HOSTNAME)
    options.add(PORT)
    options
  }

  override def optionalOptions(): util.Set[ConfigOption[_]] = {
    val options = new util.HashSet[ConfigOption[_]]()
    options.add(LOOKUP_CACHE_TTL)
    options.add(LOOKUP_CACHE_MAX_ROWS)
    options
  }

  override def createDynamicTableSource(context: DynamicTableFactory.Context): DynamicTableSource = {
    // either implement your custom validation logic here ...
    // or use the provided helper utility
    val helper = FactoryUtil.createTableFactoryHelper(this, context)

    // discover a suitable decoding format
//    val decodingFormat: DecodingFormat[DeserializationSchema[RowData]] = helper.discoverDecodingFormat(Class[DeserializationFormatFactory], FactoryUtil.FORMAT)

    // validate all options
    helper.validate()

    // get the validated options
    val options = helper.getOptions

    val configuration = context.getConfiguration

    new RedisDynamicTableSource()


  }

  override def createDynamicTableSink(context: DynamicTableFactory.Context): DynamicTableSink = ???
}
